#include <fcntl.h>
#include <google/protobuf/text_format.h>
#include <boost/scoped_array.hpp>
#include <boost/thread.hpp>
#include "Network.h"
#include "proto/operations.pb.h"
#include "proto/internal.pb.h"
#include "proto/ProtoManager.h"
#include "utils/logger.h"
#include "common.h"
#include "Op.h"
#include "worker.h"

using std::vector;

void initialize(internal::TaskList** tasks, internal::InputOperation** resultTypes) {
	if (network->my_node_number() == 0) {
		LOG("Waiting for proto.");

		std::size_t len;
		boost::scoped_array<char> proto(network->readRaw(&len));
		std::string protoString;
		protoString.append(proto.get(), len);
		LOG("Read proto (%d bytes):\n%s", len, protoString.c_str());

		LOG("Converting proto into internal protos.");
		Operation op;
		google::protobuf::TextFormat::ParseFromString(protoString, &op);
		ProtoManager protoManager(op, network->nodes_count());
		vector<internal::TaskList*> stripes = protoManager.getStripes();

		CHECK(stripes.size() == network->nodes_count(), "");
		*resultTypes = protoManager.getResultTypes(op);

		LOG("Sending internal protos.");
		for (int i = 1; i < network->nodes_count(); ++i) {
			std::string data;
			google::protobuf::TextFormat::PrintToString(*stripes[i], &data);
			delete stripes[i];
			// sending proto to thread 0, as it will have other network input on first operation (scan)
			network->sendToThread(i, 0, data.c_str(), data.length() + 1, INTERNAL_PROTO_FLAG);
		}

		*tasks = stripes[0];

		std::string data;
		google::protobuf::TextFormat::PrintToString(**tasks, &data);
		LOG("Internal proto:\n%s", data.c_str());

		network->startNetworkThread();
	} else {
		network->startNetworkThread();

		LOG("Waiting for internal proto.");
		std::size_t len;
		char flag;
		// reading proto as thread 0, as it will have other network input on first operation (scan)
		boost::scoped_array<char> internalProto(network->readAsThread(0, &len, &flag));
		CHECK(flag == INTERNAL_PROTO_FLAG, "Expected internal proto packet");
		*tasks = new internal::TaskList;
		google::protobuf::TextFormat::ParseFromString(std::string(internalProto.get()), *tasks);
		LOG("Read internal protos (%d bytes):\n%s", len, internalProto.get());
	}
}


int main(int argc, char* argv[]) {
	GOOGLE_PROTOBUF_VERIFY_VERSION;

	network = new Network(argc, argv);

	LOG("Network initialized. Initializing task list.");
	
	internal::TaskList* tasks = 0;
	internal::InputOperation* resultTypes = 0;

	initialize(&tasks, &resultTypes);

	LOG("Worker %d out of %d started.", network->my_node_number(), network->nodes_count());

	// thread 0 does the scan operation, later thread 1, 0, 1... and later on worker 0 sink
	boost::thread* workerThreads[2];
	for (int i = 0; i < 2; ++i)
		workerThreads[i] = new boost::thread(workerThread, tasks, i);

	int firstFreeThread = tasks->task_item_size() % 2;
	workerThreads[firstFreeThread]->join();
	if (network->my_node_number() == 0) {
		boost::thread sinkThread(consumeThread, resultTypes, firstFreeThread);
		sinkThread.join();
	}
	workerThreads[1 - firstFreeThread]->join();

	for (int i = 0; i < 2; ++i) {
		delete workerThreads[i];
    workerThreads[i] = 0;
  }

	network->stopNetworkThread();
	LOG("Worker %d finished.", network->my_node_number());

	delete tasks;
	delete resultTypes;
	delete network;
	google::protobuf::ShutdownProtobufLibrary();

	return 0;
}
